package com.minelx.datamicroscope.core.condition;

import com.minelx.datamicroscope.core.query.AliasedKey;
import io.gitee.minelx.commontools.factor.CloseableIterator;
import io.gitee.minelx.commontools.factor.Factor;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.function.Supplier;

import static java.util.stream.Collectors.toList;

public class UnionKeyStreamSourceFactorFactory implements ConditionFactorFactory {
    private final List<AliasedKey> keys;
    private final Supplier<InputStream> streamSupplier;

    public UnionKeyStreamSourceFactorFactory(List<AliasedKey> keys, Supplier<InputStream> streamSupplier) {
        this.keys = keys;
        this.streamSupplier = streamSupplier;
    }

    @Override
    public Factor<Condition> factor() {
        return () -> csvContentAsUnionColumnsConditionIterator(streamSupplier.get());
    }

    private CloseableIterator<Condition> csvContentAsUnionColumnsConditionIterator(InputStream inputStream) {
        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
        try {
            List<String> title = Arrays.asList(reader.readLine().split(","));
            Iterator<Condition> conditionIterator = reader.lines()
                    .map(String::trim)
                    .map(content -> content.split(","))
                    .map(line -> lineToCondition(title, line, keys))
                    .iterator();
            return CloseableIterator.of(conditionIterator, inputStream);
        } catch (IOException e) {
            throw new RuntimeException("error while doing...", e);
        }
    }

    private static Condition lineToCondition(List<String> title, String[] line, List<AliasedKey> keys) {
        return Condition.and(keys.stream()
                .map(key -> KeyOperandCondition.of(key.getKey(), Operand.eq(line[title.indexOf(key.getAlias())])))
                .collect(toList()));
    }
}
